package cn.cstor.mailtest.mqttdemo;


import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.Ordered;
import org.springframework.core.env.Environment;

/**
 * 道闸小门打卡记录监听，MQTT协议
 * @auth yuezejian
 */
//@Component
public class MqttListener implements ApplicationRunner, Ordered {

    @Autowired
    private Environment env;

    public static final Logger LOG = LoggerFactory.getLogger(MqttListener.class);
    @Override
    public void run(ApplicationArguments args) throws Exception {
        //MQTT协议前缀 urlFrontSuffix = tcp://:@192.168.20.48:1883
        String urlFrontSuffix = env.getProperty("barriergate.urlFrontSuffix", String.class);
        String clienId = String.valueOf(System.currentTimeMillis());
        System.out.println(clienId);
        //Topic前缀  allTopic = record/ba3ec3e4-2e1f-4e8d-9ff8-e0235f6e1f41/report
        String allTopic = env.getProperty("barriergate.allTopic", String.class);
        try {
            MemoryPersistence memoryPersistence = new MemoryPersistence();
            MqttClient client = new MqttClient(urlFrontSuffix, clienId, memoryPersistence);
            MqttConnectOptions options = new MqttConnectOptions();
            // 设置客户端和服务器是否应在重新启动和重新连接期间记住状态 默认false
            options.setCleanSession(true);
            // 设置会话心跳时间
            options.setKeepAliveInterval(20);
            // 设置超时时间
            options.setConnectionTimeout(10);
            // 设置回调函数，当订阅到信息时调用此方法
            client.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable throwable) {
                    // 连接失败时调用  重新连接订阅
                    try {
                        System.out.println("开始重连");
                        Thread.sleep(3000);
                        client.connect(options);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (MqttSecurityException e) {
                        e.printStackTrace();
                    } catch (MqttException e) {
                        e.printStackTrace();
                    }
                }

                @Override
                public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                    // 订阅成功，并接受信息时调用
                    String payload = new String(mqttMessage.getPayload()); // 获取消息内容
                    JSONObject pa=JSONObject.parseObject(payload);
                    String person = pa.getString("person");
                    if (StringUtils.isNotEmpty(person)) {
                        JSONObject personObject = JSONObject.parseObject(person);
                        String jobNum = personObject.getString("trdID");
                        if (StringUtils.isNotEmpty(jobNum)) {
                            //插入数据库
                        }
                    }
                    LOG.debug("MqttAccept_________"+ payload);
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    LOG.debug("deliveryComplete");
                }
            });
            client.connect(options);
            client.subscribe(allTopic,0);
            System.out.println("连接成功");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public int getOrder() {
        return 3;
    }
}
